package com.edu.flink.table;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.typeutils.TupleTypeInfo;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;


/**
 * <p>
 * 1. 将DataStream转换成Table
 * 2. 将DataStream注册成Table
 * 3. 将Table转换成DataStream
 * </p>
 *
 * @author jpge
 * @since 2024-08-20
 */
public class TransferTable {

    public static void main(String[] args) throws Exception {
        //1.定义执行环境
        StreamExecutionEnvironment streamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment streamTableEnvironment = StreamTableEnvironment.create(streamExecutionEnvironment);

        //2.定义数据源
        DataStream<Tuple2<String, Integer>> dataStreamSource = streamExecutionEnvironment.fromElements(
                new Tuple2<>("hello", 3),
                new Tuple2<>("world", 2)
        );

        //将DataStream转换成flink table
        Table table = streamTableEnvironment.fromDataStream(dataStreamSource);

        //将DataStream注册成一个flink table
        streamTableEnvironment.registerDataStream("table1", dataStreamSource); //f0, f1...
        streamTableEnvironment.registerDataStream("table2", dataStreamSource);

        //将flink table转换成DataStream
        TupleTypeInfo<Tuple2<String, Integer>> tuple2TupleTypeInfo = new TupleTypeInfo<>(
                Types.STRING,
                Types.INT
        );
        DataStream<Tuple2<String, Integer>> dataStream = streamTableEnvironment.toAppendStream(table, tuple2TupleTypeInfo);

        dataStream.print();

        streamExecutionEnvironment.execute();
    }

}
